package io.jnet.server.read;

import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import com.jfireframework.baseutil.collection.buffer.ByteBuf;
import com.jfireframework.baseutil.concurrent.CpuCachePadingInt;
import io.jnet.common.bufstorage.BufStorage;
import io.jnet.common.decodec.DecodeResult;
import io.jnet.common.decodec.FrameDecodec;
import io.jnet.common.exception.EndOfStreamException;
import io.jnet.common.exception.NotFitProtocolException;
import io.jnet.common.streamprocessor.StreamProcessor;
import io.jnet.common.util.ProcesserUtil;
import io.jnet.server.api.ServerListener;
import io.jnet.server.api.ReadHandler;
import io.jnet.server.api.ServerChannelContext;

public class SimpleReadHandler implements ReadHandler
{
    private final FrameDecodec              frameDecodec;
    private final BufStorage                bufStorage;
    private final ByteBuf<?>                ioBuf;
    private final ServerChannelContext      serverChannelContext;
    private final AsynchronousSocketChannel socketChannel;
    private final ServerListener     channelLisntener;
    private final static int                WORK   = 1;
    private final static int                IDLE   = 2;
    private final CpuCachePadingInt         status = new CpuCachePadingInt(IDLE);
    /**
     * 上一次解析出来报文后没有空余空间放入bufStorage的处理的buf
     */
    private ByteBuf<?>                      unPutedBuf;
    private final StreamProcessor[]         processors;
    
    public SimpleReadHandler(StreamProcessor[] processors, AsynchronousSocketChannel socketChannel, BufStorage bufStorage, FrameDecodec frameDecodec, ByteBuf<?> ioBuf, ServerListener channelListener, ServerChannelContext serverChannelContext)
    {
        this.processors = processors;
        this.socketChannel = socketChannel;
        this.bufStorage = bufStorage;
        this.frameDecodec = frameDecodec;
        this.ioBuf = ioBuf;
        this.channelLisntener = channelListener;
        this.serverChannelContext = serverChannelContext;
    }
    
    @Override
    public void completed(Integer read, Void nothing)
    {
        if (read == -1)
        {
            catchThrowable(EndOfStreamException.instance, serverChannelContext);
            return;
        }
        ioBuf.addWriteIndex(read);
        try
        {
            decodecAndProcess();
            socketChannel.read(getWriteBuffer(), null, this);
        }
        catch (Throwable e)
        {
            catchThrowable(e, serverChannelContext);
        }
    }
    
    @Override
    public void failed(Throwable exc, Void nothing)
    {
        catchThrowable(exc, serverChannelContext);
    }
    
    public void decodecAndProcess() throws Throwable
    {
        while (true)
        {
            DecodeResult decodeResult = frameDecodec.decodec(ioBuf);
            switch (decodeResult.getType())
            {
                case LESS_THAN_PROTOCOL:
                    return;
                case BUF_NOT_ENOUGH:
                    ioBuf.compact().ensureCapacity(decodeResult.getNeed());
                    return;
                case NOT_FIT_PROTOCOL:
                    catchThrowable(NotFitProtocolException.instance, serverChannelContext);
                    return;
                case NORMAL:
                    ByteBuf<?> packet = decodeResult.getBuf();
                    if (process(packet) == false)
                    {
                        backToIdle(serverChannelContext);
                        return;
                    }
            }
        }
    }
    
    /**
     * 将iobuf的内容进行压缩，返回一个处于可写状态的ByteBuffer
     * 
     * @return
     */
    private ByteBuffer getWriteBuffer()
    {
        ByteBuffer ioBuffer = ioBuf.nioBuffer();
        ioBuffer.position(ioBuffer.limit()).limit(ioBuffer.capacity());
        return ioBuffer;
    }
    
    private void catchThrowable(Throwable e, ServerChannelContext context)
    {
        channelLisntener.catchException(e, context);
    }
    
    private boolean process(ByteBuf<?> buf) throws Throwable
    {
        Object finalResult = ProcesserUtil.process(serverChannelContext, processors, buf, 0);
        if (finalResult instanceof ByteBuf<?>)
        {
            if (bufStorage.putBuf(buf))
            {
                serverChannelContext.commitWrite();
                return true;
            }
            else
            {
                return false;
            }
        }
        return true;
    }
    
    private void backToIdle(ServerChannelContext serverChannelContext)
    {
        status.set(IDLE);
        if (bufStorage.canPutMore())
        {
            registerRead();
        }
        else
        {
            channelLisntener.backToIdle(serverChannelContext);
        }
    }
    
    public void registerRead()
    {
        int now = status.value();
        if (now == IDLE && status.compareAndSwap(IDLE, WORK))
        {
            if (unPutedBuf == null)
            {
                channelLisntener.readRegister(serverChannelContext);
                socketChannel.read(getWriteBuffer(), null, this);
                return;
            }
            try
            {
                if (bufStorage.putBuf(unPutedBuf) == false)
                {
                    backToIdle(serverChannelContext);
                    return;
                }
                // 处理完毕
                unPutedBuf = null;
                serverChannelContext.commitWrite();
                channelLisntener.readRegister(serverChannelContext);
                socketChannel.read(getWriteBuffer(), null, this);
            }
            catch (Throwable e)
            {
                catchThrowable(e, serverChannelContext);
            }
        }
    }
    
}
